﻿using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Collections;
using System.Text;

namespace Es
{
    public class ZkClient : IZkClient
    {
        //---------------------------------------------------------------------
        ZkConnection mConnection;
        bool mShutdownTriggered;
        volatile int mEventId = 0;
        List<FuncWatchHandler> mListHandler = new List<FuncWatchHandler>();
        ConcurrentDictionary<string, zkHandlerParam> mListenerChild = new ConcurrentDictionary<string, zkHandlerParam>();
        ConcurrentDictionary<string, zkHandlerParam> mListenerData = new ConcurrentDictionary<string, zkHandlerParam>();
        ConcurrentDictionary<string, zkHandlerParam> mListenerExists = new ConcurrentDictionary<string, zkHandlerParam>();
        ConcurrentQueue<ResponseEvent> mQueEvent = new ConcurrentQueue<ResponseEvent>();
        ConcurrentQueue<WatchedEvent> mQueWatchedEvent = new ConcurrentQueue<WatchedEvent>();
        ConcurrentDictionary<int, zkHandlerParam> mMapHandler = new ConcurrentDictionary<int, zkHandlerParam>();
        zookeeper.watcher_fn mFuncWatcher;
        zookeeper.string_completion_t mFuncCreate;
        zookeeper.void_completion_t mFuncDelete;
        zookeeper.stat_completion_t mFuncExists;
        zookeeper.strings_completion_t mFuncChild;
        zookeeper.data_completion_t mFuncReaddata;
        zookeeper.stat_completion_t mFuncWritedata;

        //---------------------------------------------------------------------
        public ILog Log { get; private set; }

        //---------------------------------------------------------------------
        public ZkClient(string serverstring, string repositoryName = "")
            : this(serverstring, int.MaxValue >> 2, repositoryName)
        {
        }

        //---------------------------------------------------------------------
        public ZkClient(string zkServers, int connectionTimeout, string repositoryName)
            : this(new ZkConnection(zkServers, connectionTimeout), connectionTimeout, repositoryName)
        {
        }

        //---------------------------------------------------------------------
        public ZkClient(ZkConnection zkConnection, int connectionTimeout, string repositoryName)
        {
            // 独立日志.
            if (string.IsNullOrEmpty(repositoryName))
            {
                Log = EbLogFactory.GetLog(this.GetType().Name);
            }
            else
            {
                Log = EbLogFactory.GetLog(repositoryName, this.GetType().Name);
            }

            // 传给zk底层的调用函数，可能会被GC认为是没有引用的部分而被释放
            // 强制设置GC不去处理.
            GC.KeepAlive(mFuncWatcher);
            GC.KeepAlive(mFuncCreate);
            GC.KeepAlive(mFuncDelete);
            GC.KeepAlive(mFuncExists);
            GC.KeepAlive(mFuncChild);
            GC.KeepAlive(mFuncReaddata);
            GC.KeepAlive(mFuncWritedata);

            mFuncWatcher = _funcWatch;
            mFuncCreate = _createCompletion;
            mFuncDelete = _deleteCompletion;
            mFuncExists = _existsCompletion;
            mFuncChild = _getChildrenCompletion;
            mFuncReaddata = _getCompletion;
            mFuncWritedata = _setCompletion;

            ZK_CONST.Init();
            mConnection = zkConnection;
            _connect(connectionTimeout, mFuncWatcher);
        }

        //---------------------------------------------------------------------
        public void addWatchHandler(FuncWatchHandler handler)
        {
            mListHandler.Add(handler);
        }

        //---------------------------------------------------------------------
        static public void setlogdir(string path, int lvl)
        {
            ZkConnection.setlogdir(path, lvl);
        }

        //---------------------------------------------------------------------
        public void close()
        {
            if (mConnection == null)
            {
                return;
            }
            setShutdownTrigger(true);
            mConnection.close();
            mConnection = null;
            ZK_CONST.Release();

            GC.SuppressFinalize(mFuncWatcher);
            GC.SuppressFinalize(mFuncCreate);
            GC.SuppressFinalize(mFuncDelete);
            GC.SuppressFinalize(mFuncExists);
            GC.SuppressFinalize(mFuncChild);
            GC.SuppressFinalize(mFuncReaddata);
            GC.SuppressFinalize(mFuncWritedata);
        }

        //---------------------------------------------------------------------
        public void update()
        {
            // 处理process消息.
            while (mQueEvent.Count > 0)
            {
                ResponseEvent response;
                mQueEvent.TryDequeue(out response);

                if (null == response) continue;

                if (response.id <= 0)
                {
                    continue;
                }
                zkHandlerParam hp = null;
                mMapHandler.TryRemove(response.id, out hp);
                if (null == hp || null == hp.handler)
                {
                    Log.Error("zk update error , hp or handler is null !");
                    continue;
                }
                int eventId = mEventId++;
                switch ((ZOO_OPE)response.opeType)
                {
                    case ZOO_OPE.ZOO_CREATE_OP:
                        hp.handler(response.returnCode, response.data, null, hp.param);
                        break;
                    case ZOO_OPE.ZOO_DELETE_OP:
                        hp.handler(response.returnCode, null, null, hp.param);
                        break;
                    case ZOO_OPE.ZOO_EXISTS_OP:
                        hp.handler(response.returnCode, null, null, hp.param);
                        break;
                    case ZOO_OPE.ZOO_GETDATA_OP:
                        hp.handler(response.returnCode, response.data, null, hp.param);
                        break;
                    case ZOO_OPE.ZOO_SETDATA_OP:
                        hp.handler(response.returnCode, null, null, hp.param);
                        break;
                    case ZOO_OPE.ZOO_GETCHILDREN_OP:
                        hp.handler(response.returnCode, null, response.children, hp.param);
                        break;
                    default:
                        Log.Error("zookeeper update response.opeType error !" + response.opeType);
                        break;
                }
            }

            while (mQueWatchedEvent.Count > 0)
            {
                WatchedEvent wcEvent;
                mQueWatchedEvent.TryDequeue(out wcEvent);
                if (null == wcEvent) continue;

                string path = wcEvent.Path;
                if (mListenerChild.ContainsKey(path))
                {
                    zkHandlerParam hp;
                    mListenerChild.TryGetValue(path, out hp);
                    if (hp != null)
                    {
                        asyncGetChildren(path, true, hp.handler, hp.param);
                    }
                    else
                    {
                        asyncGetChildren(path, true, null);
                    }
                }

                if (mListenerData.ContainsKey(path))
                {
                    zkHandlerParam hp;
                    mListenerData.TryGetValue(path, out hp);
                    if (hp != null)
                    {
                        asyncReadData(path, true, hp.handler, hp.param);
                    }
                    else
                    {
                        asyncReadData(path, true, null);
                    }
                }

                if (mListenerExists.ContainsKey(path))
                {
                    zkHandlerParam hp;
                    mListenerExists.TryGetValue(path, out hp);
                    if (hp != null)
                    {
                        asyncExists(path, true, hp.handler, hp.param);
                    }
                    else
                    {
                        asyncExists(path, true, null, null);
                    }
                }

                int eventId = mEventId++;
                foreach (var func in mListHandler)
                {
                    func(wcEvent);
                }
            }
        }

        //---------------------------------------------------------------------
        public int asyncCreate(string path, string data, int mode, zkOpeHandler handler = null, Dictionary<string, object> param = null)
        {
            if (path == null || data == null)
            {
                Log.Error("path must not be null.");
                return -1;
            }

            int id = _addResponseHandler(handler, param);
            return mConnection.acreate(path, data, mode, mFuncCreate, id);
        }

        //---------------------------------------------------------------------
        public int asyncDelete(string path, zkOpeHandler handler = null, Dictionary<string, object> param = null)
        {
            int id = _addResponseHandler(handler, param);
            return mConnection.adelete(path, mFuncDelete, id);
        }

        //---------------------------------------------------------------------
        public int asyncExists(string path, bool watch, zkOpeHandler handler = null, Dictionary<string, object> param = null)
        {
            int id = _addResponseHandler(handler, param);
            return mConnection.aexists(path, watch, mFuncExists, id);
        }

        //---------------------------------------------------------------------
        public int asyncReadData(string path, bool watch, zkOpeHandler handler = null, Dictionary<string, object> param = null)
        {
            int id = _addResponseHandler(handler, param);
            return mConnection.areadData(path, watch, mFuncReaddata, id);
        }

        //---------------------------------------------------------------------
        public int asyncWriteData(string path, string data, zkOpeHandler handler = null, Dictionary<string, object> param = null)
        {
            int id = _addResponseHandler(handler, param);
            return mConnection.awriteData(path, data, mFuncWritedata, id);
        }

        //---------------------------------------------------------------------
        public int asyncGetChildren(string path, bool watch, zkOpeHandler handler = null, Dictionary<string, object> param = null)
        {
            int id = _addResponseHandler(handler, param);
            return mConnection.agetChildren(path, watch, mFuncChild, id);
        }

        //---------------------------------------------------------------------
        public void subscribeChildChanges(string path, zkOpeHandler handler = null, Dictionary<string, object> param = null)
        {
            if (path != null && path != "" && !mListenerChild.ContainsKey(path))
            {
                zkHandlerParam hp = new zkHandlerParam();
                hp.handler = handler;
                hp.param = param;
                mListenerChild.TryAdd(path, hp);
                asyncGetChildren(path, true, handler, param);
            }
        }

        //---------------------------------------------------------------------
        public void unsubscribeChildChanges(string path)
        {
            if (mListenerChild.ContainsKey(path))
            {
                zkHandlerParam rm;
                mListenerChild.TryRemove(path, out rm);
            }
        }

        //---------------------------------------------------------------------
        public void subscribeDataChanges(string path, zkOpeHandler handler = null, Dictionary<string, object> param = null)
        {
            if (path != null && path != "" && !mListenerData.ContainsKey(path))
            {
                zkHandlerParam hp = new zkHandlerParam();
                hp.handler = handler;
                hp.param = param;
                mListenerData.TryAdd(path, hp);
                asyncReadData(path, true, handler, param);
            }
        }

        //---------------------------------------------------------------------
        public void unsubscribeDataChanges(string path)
        {
            if (mListenerData.ContainsKey(path))
            {
                zkHandlerParam rm;
                mListenerData.TryRemove(path, out rm);
            }
        }

        //---------------------------------------------------------------------
        public void subscribeExists(string path, zkOpeHandler handler = null, Dictionary<string, object> param = null)
        {
            if (path != null && path != "" && !mListenerExists.ContainsKey(path))
            {
                zkHandlerParam hp = new zkHandlerParam();
                hp.handler = handler;
                hp.param = param;
                mListenerExists.TryAdd(path, hp);
                asyncExists(path, true, handler, param);
            }
        }

        //---------------------------------------------------------------------
        public void unsubscribeExists(string path)
        {
            if (mListenerExists.ContainsKey(path))
            {
                zkHandlerParam rm;
                mListenerExists.TryRemove(path, out rm);
            }
        }

        //---------------------------------------------------------------------
        public string syncCreate(string path, string value, int flags)
        {
            string outStr;
            int rt = mConnection.screate(path, value, flags, out outStr);
            return outStr;
        }

        //---------------------------------------------------------------------
        public bool syncDelete(string path)
        {
            return mConnection.sdelete(path) == 0;
        }

        //---------------------------------------------------------------------
        public bool syncExists(string path, bool watch)
        {
            return mConnection.sexists(path, watch);
        }

        //---------------------------------------------------------------------
        public string syncRead(string path, bool watch)
        {
            return mConnection.sread(path, watch);
        }

        //---------------------------------------------------------------------
        public int syncWrite(string path, string buffer)
        {
            return mConnection.swrite(path, buffer);
        }

        //---------------------------------------------------------------------
        public string[] syncGetChildren(string path, bool watch)
        {
            return mConnection.sget_children(path, watch);
        }

        //---------------------------------------------------------------------
        public void Dispose()
        {
            close();

            ZK_CONST.Release();
        }

        //---------------------------------------------------------------------
        public int getEventCount()
        {
            return (mQueEvent.Count + mQueWatchedEvent.Count);
        }

        //---------------------------------------------------------------------
        public void setShutdownTrigger(bool triggerState)
        {
            mShutdownTriggered = triggerState;
        }

        //---------------------------------------------------------------------
        public bool getShutdownTrigger()
        {
            return mShutdownTriggered;
        }

        //---------------------------------------------------------------------
        public int numberOfListeners()
        {
            return (mListenerChild.Count + mListenerData.Count + mListenerExists.Count);
        }

        //---------------------------------------------------------------------
        public int getState()
        {
            return mConnection.getState();
        }

        //---------------------------------------------------------------------
        public void dumpWatchPath()
        {
            Log.Info("watch for children, count:" + mListenerChild.Count);
            int index = 0;
            foreach (var child in mListenerChild)
            {
                Log.Info("index #" + index + ":" + child.Key);
            }

            index = 0;
            Log.Info("watch for data, count:" + mListenerData.Count);
            foreach (var data in mListenerData)
            {
                Log.Info("index #" + index + ":" + data.Key);
            }

            index = 0;
            Log.Info("watch for exists, count:" + mListenerExists.Count);
            foreach (var data in mListenerExists)
            {
                Log.Info("index #" + index + ":" + data.Key);
            }
        }

        //---------------------------------------------------------------------
        void _funcWatch(IntPtr zh, int type, int state, IntPtr ptrpath, IntPtr watcherCtx)
        {
            if (type == (int)ZOO_EVENT.SESSION_EVENT_DEF)
            {
                if (state == (int)ZOO_STATE.CONNECTED_STATE_DEF)
                {
                    //mLog.Info("ZkClient connected successfully!");
                }
                else if (state == (int)ZOO_STATE.EXPIRED_SESSION_STATE_DEF)
                {
                    //mLog.Error("ZkClient disconnected!");
                }
                return;
            }

            if (type == (int)ZOO_EVENT.NOTWATCHING_EVENT_DEF)
            {
                //mLog.Error("ZOO_EVENT.NOTWATCHING_EVENT_DEF");
                //return;
            }

            string path = zookeeper.getString(ptrpath);
            if (path == null || path == "")
            {
                //mLog.Error("watchFunction path is null !");
                //return;
            }

            WatchedEvent ev = new WatchedEvent(state, type, path);
            mQueWatchedEvent.Enqueue(ev);
        }

        //---------------------------------------------------------------------
        void _connect(int maxMsToWaitUntilConnected, zookeeper.watcher_fn watcher)
        {
            setShutdownTrigger(false);
            mConnection.connect(watcher);
        }

        //---------------------------------------------------------------------
        void _deleteCompletion(int rc, IntPtr data)
        {
            int id = zookeeper.getInt(data);
            if (data != IntPtr.Zero) zookeeper.FreeMem(data); data = IntPtr.Zero;
            {
                mQueEvent.Enqueue(new ResponseEvent((int)ZOO_OPE.ZOO_DELETE_OP, rc, null, id, null));
            }
        }

        //---------------------------------------------------------------------
        void _existsCompletion(int rc, IntPtr stat, IntPtr data)
        {
            int id = zookeeper.getInt(data);
            if (data != IntPtr.Zero) zookeeper.FreeMem(data); data = IntPtr.Zero;
            if (rc != 0)
            {

            }
            mQueEvent.Enqueue(new ResponseEvent((int)ZOO_OPE.ZOO_EXISTS_OP, rc, null, id, null));
        }

        //---------------------------------------------------------------------
        void _setCompletion(int rc, IntPtr stat, IntPtr data)
        {
            int id = zookeeper.getInt(data);
            if (data != IntPtr.Zero) zookeeper.FreeMem(data); data = IntPtr.Zero;
            mQueEvent.Enqueue(new ResponseEvent((int)ZOO_OPE.ZOO_SETDATA_OP, rc, null, id, null));
        }

        //---------------------------------------------------------------------
        void _createCompletion(int rc, IntPtr value, IntPtr data)
        {
            int id = zookeeper.getInt(data);
            if (data != IntPtr.Zero) zookeeper.FreeMem(data); data = IntPtr.Zero;
            if (rc != 0)
            {
                mQueEvent.Enqueue(new ResponseEvent((int)ZOO_OPE.ZOO_CREATE_OP, rc, null, id, null));
            }
            else
            {
                mQueEvent.Enqueue(new ResponseEvent((int)ZOO_OPE.ZOO_CREATE_OP, rc, zookeeper.getString(value), id, null));
            }
        }

        //---------------------------------------------------------------------
        void _getCompletion(int rc, IntPtr value, int value_len, IntPtr stat, IntPtr data)
        {
            int id = zookeeper.getInt(data);
            if (data != IntPtr.Zero) zookeeper.FreeMem(data); data = IntPtr.Zero;
            if (rc != 0)
            {
                mQueEvent.Enqueue(new ResponseEvent((int)ZOO_OPE.ZOO_GETDATA_OP, rc, null, id, null));
            }
            else
            {
                string val = zookeeper.getString2(value, value_len);
                mQueEvent.Enqueue(new ResponseEvent((int)ZOO_OPE.ZOO_GETDATA_OP, rc, val, id, null));
            }
        }

        //---------------------------------------------------------------------
        void _getChildrenCompletion(int rc, IntPtr strings, IntPtr data)
        {
            int id = zookeeper.getInt(data);
            if (data != IntPtr.Zero) zookeeper.FreeMem(data); data = IntPtr.Zero;
            if (rc != 0)
            {
                mQueEvent.Enqueue(new ResponseEvent((int)ZOO_OPE.ZOO_GETCHILDREN_OP, rc, null, id, null));
            }
            else
            {
                mQueEvent.Enqueue(new ResponseEvent((int)ZOO_OPE.ZOO_GETCHILDREN_OP, rc,
                    null, id, zookeeper.getString_vector(strings)));
            }
        }

        //---------------------------------------------------------------------
        int _addResponseHandler(zkOpeHandler handler, Dictionary<string, object> param)
        {
            int id = 0;
            if (handler != null)
            {
                id = zookeeper.generateId();
                zkHandlerParam hp = new zkHandlerParam();
                hp.handler = handler;
                hp.param = param;
                mMapHandler.TryAdd(id, hp);
            }
            return id;
        }
    }
}
